package com.xnf.article.stream.listener;

import com.alibaba.fastjson.JSON;
import com.xnf.article.config.stream.KafkaStreamListener;
import com.xnf.article.service.ApArticleService;
import com.xnf.common.message.HotArticleConstants;
import com.xnf.model.article.mess.ArticleVisitStreamMess;
import com.xnf.model.mess.app.UpdateArticleMess;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.Arrays;

/**
 * Created by chendx on 2022/4/6.
 */
@Component
@Slf4j
public class HotArticleStreamHandler implements KafkaStreamListener<KStream<String,String>> {

    @Autowired
    private ApArticleService apArticleService;

    @Override
    public String listenerTopic() {
        return HotArticleConstants.HOT_ARTICLE_SCORE_TOPIC;
    }

    @Override
    public String sendTopic() {
        return HotArticleConstants.HOT_ARTICLE_INCR_HANDLE_TOPIC;
    }

    @Override
    public KStream<String, String> getService(KStream<String, String> stream) {
        return stream.flatMapValues(new ValueMapper<String, Iterable<String>>() {
            @Override
            public Iterable<String> apply(String value) {
                UpdateArticleMess updateArticleMess = JSON.parseObject(value, UpdateArticleMess.class);
                log.info("recevie message value:{}",updateArticleMess);
                System.out.println(updateArticleMess);
                //按照文章的id和行为进行聚合查询  --》  1232323232323:LIEKS
              //  ResponseResult responseResult = apArticleService.updateApArticleLVC(updateArticleMess.getType().name(), updateArticleMess.getArticleId());
              //  System.out.println(responseResult);
                return Arrays.asList(updateArticleMess.getArticleId()+":"+updateArticleMess.getType());
            }
        }).map(new KeyValueMapper<String, String, KeyValue<String,String>>() {
            @Override
            public KeyValue<String, String> apply(String key, String value) {
                //key
                //value = 1232323232323:LIEKS
                System.out.println(key +" "+value);
                return new KeyValue<String, String>(value,value);
            }
        }).groupByKey().windowedBy(TimeWindows.of(10000))
                .count(Materialized.as("count-article-num-000001"))
                .toStream().map(new KeyValueMapper<Windowed<String>, Long, KeyValue<String, String>>() {

                    @Override
                    public KeyValue<String, String> apply(Windowed<String> key, Long value) {
                        //key.key() =>1232323232323:LIEKS
                        //value => count()
                        System.out.println(key +" map "+value);
                        ArticleVisitStreamMess mess = new ArticleVisitStreamMess();
                        String mapKey = key.key();
                        String[] split = mapKey.split(":");
                        mess.setArticleId(Long.valueOf(split[0]));
                        if(split[1].equals(UpdateArticleMess.UpdateArticleType.LIKES.name())){
                            mess.setLike(Long.valueOf(value));
                        }
                        if(split[1].equals(UpdateArticleMess.UpdateArticleType.VIEWS.name())){
                            mess.setView(Long.valueOf(value));
                        }
                        if(split[1].equals(UpdateArticleMess.UpdateArticleType.COLLECTION.name())){
                            mess.setCollect(Long.valueOf(value));
                        }
                        if(split[1].equals(UpdateArticleMess.UpdateArticleType.COMMENT.name())){
                            mess.setComment(Long.valueOf(value));
                        }
                        log.info("处理之后的结果为:{}",mess);
                        System.out.println("处理之后的结果为 " + mess);
                        return new KeyValue<>(key.key(),JSON.toJSONString(mess));
                    }
                });
    }
}
